package bx.job.httpserver.route;

import akka.actor.typed.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.Scheduler;
import akka.actor.typed.javadsl.AskPattern;
import akka.http.javadsl.marshallers.jackson.Jackson;
import akka.http.javadsl.model.ContentTypes;
import akka.http.javadsl.model.HttpResponse;
import akka.http.javadsl.model.StatusCodes;
import akka.http.javadsl.server.Route;
import bx.InteractionEntity.CommonResponse;
import bx.InteractionEntity.WorkerCompleteJobRequest;
import bx.InteractionEntity.WorkerHeartbeatRequest;
import bx.InteractionEntity.WorkerRegistRequest;
import bx.job.httpserver.actor.JobWorkerHeartbeatProcessorActor;
import bx.job.httpserver.actor.JobWorkerRegistProcessorActor;
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import static akka.http.javadsl.server.Directives.*;
import static bx.InteractionEntity.CommonResponse.FAILED_STATUS;
import static bx.InteractionEntity.CommonResponse.SUCCESS_STATUS;

public class JobServerServerRoutes {
    //#user-routes-class
    private final static Logger log = LoggerFactory.getLogger(JobServerServerRoutes.class);
    private final ActorRef<JobWorkerRegistProcessorActor.JobWorkerRegistProcessorActorMsg> jobWorkerRegistProcessorActor;

    private final ActorRef<JobWorkerHeartbeatProcessorActor.JobWorkerHeartbeatProcessorActorMsg> jobWorkerHeartbeatProcessorActor;
    private final Duration askTimeout;
    private final Scheduler scheduler;

    public JobServerServerRoutes(ActorSystem<?> system, ActorRef<JobWorkerRegistProcessorActor.JobWorkerRegistProcessorActorMsg> jobWorkerRegistProcessorActor
            , ActorRef<JobWorkerHeartbeatProcessorActor.JobWorkerHeartbeatProcessorActorMsg> jobWorkerHeartbeatProcessorActor) {
        scheduler = system.scheduler();
        askTimeout = system.settings().config().getDuration("bx-job-server.routes.ask-timeout");
        this.jobWorkerRegistProcessorActor = jobWorkerRegistProcessorActor;
        this.jobWorkerHeartbeatProcessorActor = jobWorkerHeartbeatProcessorActor;
    }


    public Route jobServerServerRoutes() {
        return concat(
                path("regist", () -> post(() ->
                        entity(Jackson.unmarshaller(WorkerRegistRequest.class),
                                // 这里不要用 onSuccess，以便于处理超时异常
                                workerRegistRequest -> onComplete(toRegist(workerRegistRequest), cf -> {
                                    Gson gson = new Gson();
                                    if (cf.isFailure()) {
                                        log.error("regist worker failed: {}", workerRegistRequest, cf.failed().get());
                                        CommonResponse<Void> response = new CommonResponse<>();
                                        response.setStatus(FAILED_STATUS);
                                        response.setMsg("regist failed , please view server log");
                                        HttpResponse httpResponse = HttpResponse.create().withStatus(StatusCodes.OK).withEntity(ContentTypes.APPLICATION_JSON, gson.toJson(response));
                                        return complete(httpResponse);
                                    }
                                    log.info("regist worker success: {}", workerRegistRequest);
                                    HttpResponse httpResponse = HttpResponse.create().withStatus(StatusCodes.OK).withEntity(ContentTypes.APPLICATION_JSON, gson.toJson(cf.get()));
                                    return complete(httpResponse);
                                })))),
                path("heartbeat", () -> post(() -> entity(Jackson.unmarshaller(WorkerHeartbeatRequest.class), workerHeartbeatRequest -> onSuccess(toHeartbeat(workerHeartbeatRequest), performed -> {
                    CommonResponse<Void> response = new CommonResponse<>();
                    response.setStatus(SUCCESS_STATUS);
                    return complete(StatusCodes.OK, response, Jackson.marshaller());
                })))),
                path("jobComplete", () -> post(() -> entity(Jackson.unmarshaller(WorkerCompleteJobRequest.class), workerCompleteJobRequest -> onSuccess(toComplete(workerCompleteJobRequest), performed -> {
                    CommonResponse<Void> response = new CommonResponse<>();
                    response.setStatus(SUCCESS_STATUS);
                    return complete(StatusCodes.OK, response, Jackson.marshaller());
                }))))

        );
    }

    private CompletionStage<Object> toComplete(WorkerCompleteJobRequest workerCompleteJobRequest) {
        log.info("receive worker jobComplete request: {}", workerCompleteJobRequest);
        return null;
    }

    private CompletableFuture<String> toHeartbeat(WorkerHeartbeatRequest workerHeartbeatRequest) {
//        return AskPattern.ask(jobWorkerRegistProcessorActor, ref -> new UserRegistry.DeleteUser(name, ref), askTimeout, scheduler);
        log.info("receive worker heartbeat request: {}", workerHeartbeatRequest);
        // TODO 调用 JobWorkerHeartbeatProcessorActor 调用数据库完成心跳
        return CompletableFuture.completedFuture("ok");
    }

    // 使用 ack 时，要注意处理 timeout 超时处理
    private CompletionStage<JobWorkerRegistProcessorActor.JobWorkerRegistInfo> toRegist(WorkerRegistRequest registRequest) {
        log.info("receive worker regist request: {}", registRequest);
        return AskPattern.ask(jobWorkerRegistProcessorActor,
                actorRef -> new JobWorkerRegistProcessorActor.JobWorkerRegistMsg(registRequest, actorRef),
                askTimeout, scheduler);

    }


}
